package per.chao.mqtt.domain.repo;

import com.google.common.collect.Maps;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import per.chao.mqtt.domain.facade.IMessageCache;
import per.chao.mqtt.domain.interfaces.repo.BaseRepo;
import per.chao.mqtt.domain.shared.store.RetainMessageStore;
import per.chao.mqtt.domain.util.TopicUtil;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * Description:
 *
 * @author W.Chao
 * @date 2021/6/20 14:32
 **/
@Component
public class RetainMessageRepo extends BaseRepo<RetainMessageStore> {
	private static final String RETAIN_MESSAGE_PREFIX = "retain::";

	@Autowired
	@Qualifier("redisCache")
	private IMessageCache iMessageCache;

	private static String prefix(String topic) {
		return TopicUtil.prefix(RETAIN_MESSAGE_PREFIX, topic);
	}

	@Override
	public void put(String topic, RetainMessageStore retainMessageStore) {
		iMessageCache.put(prefix(topic), retainMessageStore);
	}

	@Override
	public RetainMessageStore get(String topicFilter) {
		return (RetainMessageStore) iMessageCache.get(prefix(topicFilter));
	}

	public List<RetainMessageStore> search(String topicFilter) {
		List<RetainMessageStore> retainMessageStores = new ArrayList<>();
		if (!topicFilter.contains("#") && !topicFilter.contains("+")) {
			if (iMessageCache.containsKey(prefix(topicFilter))) {
				retainMessageStores.add((RetainMessageStore) iMessageCache.get(prefix(topicFilter)));
			}
		} else {
			all().forEach((topic, val) -> {
				// 只有保存的topic片段大于topic过滤器片段才可以进行匹配
				if (StringUtils.countMatches(topic, "/") >= StringUtils.countMatches(topicFilter, "/")) {
					String[] splitTopics = topic.split("/");
					String[] splitTopicFilters = topicFilter.split("/");
					StringBuilder newTopicFilters = new StringBuilder();
					for (int i = 0; i < splitTopicFilters.length; i++) {
						String value = splitTopicFilters[i];
						if ("+".equals(value)) {
							newTopicFilters.append("+/");
						} else if ("#".equals(value)) {
							newTopicFilters.append("#/");
							break;
						} else {
							newTopicFilters.append(splitTopics[i]).append("/");
						}
					}
					newTopicFilters.deleteCharAt(newTopicFilters.length() - 1);
					if (topicFilter.equals(newTopicFilters.toString())) {
						retainMessageStores.add(val);
					}
				}
			});
		}
		return retainMessageStores;
	}

	@Override
	public void remove(String topic) {
		iMessageCache.remove(prefix(topic));
	}

	@Override
	public boolean containsKey(String topic) {
		return iMessageCache.containsKey(prefix(topic));
	}

	public Map<String, RetainMessageStore> all() {
		Map<String, RetainMessageStore> retainMsgMap = Maps.newHashMap();
		Set<String> keys = iMessageCache.keys(RETAIN_MESSAGE_PREFIX);
		if (keys != null && keys.size() > 0) {
			retainMsgMap = keys.stream()
					.collect(Collectors.toMap(key -> key.substring(RETAIN_MESSAGE_PREFIX.length()),
							key -> (RetainMessageStore) iMessageCache.get(key)));
		}
		return retainMsgMap;
	}
}
